//===----------------------------------------------------------------------===//
//
//                         BusTub
//
// buffer_pool_manager.cpp
//
// Identification: src/buffer/buffer_pool_manager.cpp
//
// Copyright (c) 2015-2021, Carnegie Mellon University Database Group
//
//===----------------------------------------------------------------------===//

#include "buffer/buffer_pool_manager.h"
#include <cstddef>
#include <mutex>

#include "common/config.h"
#include "common/exception.h"
#include "common/macros.h"
#include "storage/page/page.h"
#include "storage/page/page_guard.h"

namespace bustub {

BufferPoolManager::BufferPoolManager(size_t pool_size, DiskManager *disk_manager, size_t replacer_k,
                                     LogManager *log_manager)
    : pool_size_(pool_size), disk_manager_(disk_manager), log_manager_(log_manager), rv_page_table_(pool_size) {
  // we allocate a consecutive memory space for the buffer pool
  pages_ = new Page[pool_size_];
  replacer_ = std::make_unique<LRUKReplacer>(pool_size, replacer_k);

  // Initially, every page is in the free list.
  for (size_t i = 0; i < pool_size_; ++i) {
    free_list_.emplace_back(static_cast<int>(i));
  }

  fill(rv_page_table_.begin(), rv_page_table_.end(), INVALID_PAGE_ID);
}

BufferPoolManager::~BufferPoolManager() { delete[] pages_; }

auto BufferPoolManager::NewPage(page_id_t *page_id) -> Page * {
  std::scoped_lock<std::mutex> lock(latch_);
  page_id_t expected = next_page_id_;
  Page *page = ReadNewPage(expected, true);
  if (page != nullptr) {
    *page_id = expected;
  }
  return page;
}

auto BufferPoolManager::FetchPage(page_id_t page_id, [[maybe_unused]] AccessType access_type) -> Page * {
  std::scoped_lock<std::mutex> lock(latch_);
  frame_id_t frame_id;
  if (page_table_.count(page_id) != 0) {
    frame_id = page_table_[page_id];
    // CRITICAL :
    // otherwise other thread would unpin an already referenced READ page
    // -> evicted by other NewPage() -> REWRITE before read
    ++pages_[frame_id].pin_count_;

    replacer_->RecordAccess(frame_id);
    replacer_->SetEvictable(frame_id, false);
    return &(pages_[frame_id]);
  }
  return ReadNewPage(page_id, false);
}

auto BufferPoolManager::UnpinPage(page_id_t page_id, bool is_dirty, [[maybe_unused]] AccessType access_type) -> bool {
  std::scoped_lock<std::mutex> lock(latch_);
  if (page_table_.count(page_id) == 0 || pages_[page_table_[page_id]].GetPinCount() <= 0) {
    return false;
  }
  frame_id_t frame_id = page_table_[page_id];
  Page &page = pages_[frame_id];

  page.is_dirty_ |= is_dirty;
  if (--page.pin_count_ == 0) {
    replacer_->SetEvictable(frame_id, true);
  }

  return true;
}

auto BufferPoolManager::FlushPage(page_id_t page_id) -> bool {
  if (page_table_.count(page_id) == 0 || page_id == INVALID_PAGE_ID) {
    return false;
  }
  Page &page = pages_[page_table_[page_id]];
  disk_manager_->WritePage(page_id, page.GetData());
  page.is_dirty_ = false;
  return true;
}

void BufferPoolManager::FlushAllPages() {
  std::scoped_lock<std::mutex> lock(latch_);
  for (auto &[page_id, _] : page_table_) {
    FlushPage(page_id);
  }
}

auto BufferPoolManager::DeletePage(page_id_t page_id) -> bool {
  std::scoped_lock<std::mutex> lock(latch_);
  if (page_table_.count(page_id) == 0) {
    return true;
  }

  frame_id_t frame_id = page_table_[page_id];
  Page &page = pages_[frame_id];

  if (page.GetPinCount() != 0) {
    return false;
  }

  replacer_->Remove(frame_id);
  page.ResetMemory();
  page.is_dirty_ = false;
  page.page_id_ = INVALID_PAGE_ID;
  page_table_.erase(page_id);
  rv_page_table_[frame_id] = INVALID_PAGE_ID;

  free_list_.push_back(frame_id);
  DeallocatePage(page_id);

  return true;
}

auto BufferPoolManager::ReadNewPage(page_id_t page_id, bool is_newpage = false) -> Page * {
  frame_id_t frame_id;
  if (!free_list_.empty()) {
    frame_id = free_list_.front();
    free_list_.pop_front();
  } else if (!replacer_->Evict(&frame_id)) {
    return nullptr;
  }

  replacer_->RecordAccess(frame_id);
  replacer_->SetEvictable(frame_id, false);

  Page &page = pages_[frame_id];

  if (page.IsDirty()) {
    FlushPage(page.GetPageId());
  }
  page.ResetMemory();
  page.page_id_ = page_id;
  page.pin_count_ = 1;
  disk_manager_->ReadPage(page_id, page.GetData());

  if (is_newpage) {
    AllocatePage();
  }

  // change mappings
  if (rv_page_table_[frame_id] != INVALID_PAGE_ID) {
    page_table_.erase(rv_page_table_[frame_id]);
  }
  page_table_[page_id] = frame_id;
  rv_page_table_[frame_id] = page_id;

  return &(pages_[frame_id]);
}

auto BufferPoolManager::AllocatePage() -> page_id_t { return next_page_id_++; }

auto BufferPoolManager::FetchPageBasic(page_id_t page_id) -> BasicPageGuard {
  Page *page = FetchPage(page_id);
  if (page == nullptr) {
    return {this, nullptr};
  }
  return {this, page};
}

auto BufferPoolManager::FetchPageRead(page_id_t page_id) -> ReadPageGuard {
  Page *page = FetchPage(page_id);
  if (page == nullptr) {
    return {this, nullptr};
  }
  page->RLatch();
  return {this, page};
}

auto BufferPoolManager::FetchPageWrite(page_id_t page_id) -> WritePageGuard {
  Page *page = FetchPage(page_id);
  if (page == nullptr) {
    return {this, nullptr};
  }
  page->WLatch();
  return {this, page};
}

auto BufferPoolManager::NewPageGuarded(page_id_t *page_id) -> BasicPageGuard {
  Page *page = NewPage(page_id);
  if (page == nullptr) {
    return {this, nullptr};
  }
  return {this, page};
}

}  // namespace bustub
